package tech.xs.common.framework.util;

import lombok.SneakyThrows;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import tech.xs.common.framework.holder.SpringContextHolder;
import tech.xs.common.framework.mq.MqMessageListener;

import java.io.ByteArrayOutputStream;
import java.io.ObjectOutputStream;

public class MqUtil {
    private final static Logger log = LoggerFactory.getLogger(MqUtil.class);

    /**
     * 创建路由key匹配队列
     *
     * @param exchangeName 交换机名
     * @param queueName    队列名
     */
    public static void createKeyMatchQueue(String exchangeName, String queueName) {
        log.info("开始创建[匹配]交换机和队列 队列名:{}", exchangeName);
        RabbitAdmin rabbitAdmin = SpringContextHolder.getBean(RabbitAdmin.class);
        QueueInformation queueInfo = rabbitAdmin.getQueueInfo(queueName);
        if (queueInfo != null) {
            log.info("交换机和队列已经存在,忽略创建 队列名:{}", queueName);
            return;
        }
        Queue queue = new Queue(queueName);
        TopicExchange exchange = new TopicExchange(exchangeName);
        Binding binding = new Binding(queueName, Binding.DestinationType.QUEUE, exchange.getName(), queueName, null);
        rabbitAdmin.declareQueue(queue);
        rabbitAdmin.declareExchange(exchange);
        rabbitAdmin.declareBinding(binding);
        log.info("[匹配]创建交换机和队列完毕 队列名:{}", exchangeName);
    }


    /**
     * 创建直连队列
     *
     * @param queueName 队列名
     */
    public static void createDirectQueue(String queueName) {
        log.info("开始创建[直连]交换机和队列 队列名:{}", queueName);
        RabbitAdmin rabbitAdmin = SpringContextHolder.getBean(RabbitAdmin.class);
        QueueInformation queueInfo = rabbitAdmin.getQueueInfo(queueName);
        if (queueInfo != null) {
            log.info("交换机和队列已经存在,忽略创建 队列名:{}", queueName);
            return;
        }
        Queue queue = new Queue(queueName);
        DirectExchange exchange = new DirectExchange(queueName);
        Binding binding = new Binding(queueName, Binding.DestinationType.QUEUE, exchange.getName(), queueName, null);
        rabbitAdmin.declareQueue(queue);
        rabbitAdmin.declareExchange(exchange);
        rabbitAdmin.declareBinding(binding);
        log.info("[直连]创建交换机和队列完毕 队列名:{}", queueName);
    }

    @SneakyThrows
    public static <T> void send(String queueName, Object data) {
        RabbitTemplate rabbitTemplate = SpringContextHolder.getBean(RabbitTemplate.class);
        ByteArrayOutputStream bytesStream = new ByteArrayOutputStream();
        ObjectOutputStream objStream = new ObjectOutputStream(bytesStream);
        objStream.writeObject(data);
        bytesStream.flush();
        objStream.flush();
        rabbitTemplate.convertAndSend(queueName, queueName, bytesStream.toByteArray());
        objStream.close();
        bytesStream.close();
    }

    public static void listener(String queueName, MessageListener listener) {
        ConnectionFactory factory = SpringContextHolder.getBean(ConnectionFactory.class);
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(factory);
        container.setQueueNames(queueName);
        container.setMessageListener(listener);
        container.setAcknowledgeMode(AcknowledgeMode.NONE);
        container.start();
    }

    public static void listener(MqMessageListener listener) {
        listener(listener.getQueueName(), listener);
    }

}
